package com.shujia.base;

import com.shujia.bean.FilterName;
import com.shujia.utils.HBaseUtil;
import org.apache.hadoop.hbase.CompareOperator;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.filter.BinaryComparator;
import org.apache.hadoop.hbase.filter.PrefixFilter;
import org.apache.hadoop.hbase.filter.SingleColumnValueFilter;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.BufferedReader;
import java.io.FileReader;
import java.util.HashMap;

public class HBasePractice {


    public static void main(String[] args) throws Exception {
        //获取hbase的连接对象和数据库操作对象
        Connection connection = HBaseUtil.CONNECTION;
        Admin admin = HBaseUtil.ADMIN;

        //在HBase中创建一张成绩表
        HBaseUtil.createOneTable("scores", "info");

        //删除一张表
//        HBaseUtil.dropOneTable("scores");

        /**
         * hbase中表的rowkey的设计要遵循三大基本原则：
         * 1、唯一性
         * 2、长度不宜过长
         * 3、散列性
         * --------------------------------------------------
         * 1500101000-1000001,100
         * 1500101000-1000002,78
         * 1500101000-1000003,81
         * 1500101000-1000007,5
         * 1500101000-1000008,87
         * 1500101000-1000009,28
         */
        //创建字符缓冲输入流对象，读取成绩文件
//        BufferedReader br = new BufferedReader(new FileReader("hbase/data/score.txt"));
//        String line = null;
//        while ((line=br.readLine())!=null){
//            //向表中添加数据
//            String[] strings = line.split(",");
//            String id = strings[0];
//            String sid = strings[1];
//            String score = strings[2];
//
//            HBaseUtil.putOneDataToTable("scores",id+"-"+sid,"info","score",score);
//        }

        //获取表对象实例
        Table students = connection.getTable(TableName.valueOf("students"));
        Table scores = connection.getTable(TableName.valueOf("scores"));

        //TODO：查询文科一班学生总分排名前10的学生（输出：学号，姓名，班级，总分）结果写到hbase
        //步骤1：使用单列值过滤器过滤出文科一班的学生
        SingleColumnValueFilter singleColumnValueFilter = new SingleColumnValueFilter(Bytes.toBytes("info"), Bytes.toBytes("clazz"),
                CompareOperator.EQUAL, new BinaryComparator(Bytes.toBytes("文科一班")));
        Scan scan = new Scan();
        scan.setFilter(singleColumnValueFilter);
        ResultScanner resultScanner = students.getScanner(scan);
        //创建一个map集合存储查询的文科一班的学生信息
        HashMap<String, Integer> yiBanStudent = new HashMap<>(); // <id+"-"+name+"-"+clazz, 200>
        //1500100273,相寒安,22,女,文科一班
        //遍历hbase查询结果
        for (Result result : resultScanner) {
            //1500100273
            String id = Bytes.toString(result.getRow());
            //拿着查询到了文科一班的学生学号，去成绩表中查询该学生的所有科目成绩
            //利用行键前缀过滤器找到对应的学生6门考试成绩
            PrefixFilter prefixFilter = new PrefixFilter(Bytes.toBytes(id));
            Scan scan1 = new Scan();
            scan1.setFilter(prefixFilter);
            ResultScanner resultScanner1 = scores.getScanner(scan1); // 这个结果里就存放了该文科一班学生的6门成绩
            /**
             * 1500100273-1000001,124
             * 1500100273-1000002,44
             * 1500100273-1000003,79
             * 1500100273-1000004,10
             * 1500100273-1000005,96
             * 1500100273-1000006,65
             */
            //遍历该学生的6门成绩求和
            int sumScore = 0;
            for (Result result1 : resultScanner1) {
                int score = Integer.parseInt(Bytes.toString(result1.getValue(Bytes.toBytes("info"), Bytes.toBytes("score"))));
                sumScore += score;
            }
            String name = Bytes.toString(result.getValue(Bytes.toBytes("info"), Bytes.toBytes("name")));
            String clazz = Bytes.toString(result.getValue(Bytes.toBytes("info"), Bytes.toBytes("clazz")));

            yiBanStudent.put(id + "-" + name + "-" + clazz, sumScore);
        }

        //需求2：将计算的结果存储到hbase表中后，我们可以使用LIMIT修饰词来获取总成绩前n的同学
        /**
         * 思考：
         *   不能够直接使用学号作为rowkey
         *   我们想到可以使用总分作为rowkey, 又为了以后查询可以知道该成绩是哪一位同学的条件
         *   所以我们就想，可以将总成绩与学号一并作为rowkey存储
         *   353-1500100946  秋海白	文科一班
         *   328-1500100308  黄初夏  文科一班
         *   ...
         *  我们可以拿一个较大的值先减去分数，得到一个结果值
         *  672-1500100308  黄初夏  文科一班
         *  647-1500100946  秋海白	文科一班
         */

        HBaseUtil.dropOneTable("clazzWenOne");
        //创建结果表
        HBaseUtil.createOneTable("clazzWenOne","info");

        yiBanStudent.forEach((k, v) -> {
            String[] strings = k.split("-");
            String id = strings[0];
            String name = strings[1];
            String clazz = strings[2];
            int new_score = 1000 - v;
            System.out.println("学号:" + strings[0] + "\t姓名:" + strings[1] + "\t班级:" + strings[2] + "\t总分:" + v);
            HBaseUtil.putOneDataToTable("clazzWenOne",new_score+"-"+id,"info","name",name);
            HBaseUtil.putOneDataToTable("clazzWenOne",new_score+"-"+id,"info","clazz",clazz);
        });


        //释放资源
        HBaseUtil.closeSource();
    }
}
